package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.util.Random

/**
  * create by undeRdoG on  2021-06-19  18:02
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object SparkStreaming03_DIY {

  def main(args: Array[String]): Unit = {
    // 创建环境
    /**
      * StreamingContext  创建时需要两个参数
      * 第一个参数环境配置
      * 第二个参数 批处理周期  duration
      **/
    val sparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(3))


    /**
      * receiverStream
      **/

    val messageDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver())

    messageDS.print()


    /**
      * 由于sparkStreaming 采集器是长期执行的任务，所以不能直接关闭
      * 如果main方法执行完毕，程序也会结束
      * 因此不能让main方法执行完毕
      **/

    // 启动采集器
    ssc.start()

    // 等待采集器关闭
    ssc.awaitTermination()

  }

  /**
    * 自定义数据采集器
    * 定义泛型，传递参数
    **/
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {

    private var flag = true

    override def onStart(): Unit = {
      new Thread(new Runnable {
        override def run(): Unit = {
          while (flag) {


            val message = "采集数据为：" + new Random().nextInt(10).toString

            // 存储消息
            store(message)

            Thread.sleep(500)
          }
        }
      }).start()
    }

    override def onStop(): Unit = {
      flag = false
    }
  }

}
